package taskscheduleservicelogic

import (
	"context"
	"github.com/zeromicro/go-zero/core/service"
	"go-zero-admin/application/job/rpc/internal/config"
	"go-zero-admin/application/job/rpc/internal/logic"
	"go-zero-admin/application/job/rpc/internal/scheduler"
	"sync"
	"time"

	"go-zero-admin/application/job/rpc/internal/svc"
	job "go-zero-admin/application/job/rpc/pb"

	"github.com/zeromicro/go-zero/core/logx"
)

type ReloadCronLogic struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
	logx.Logger
	mu     sync.Mutex // 用于并发控制
	wg     sync.WaitGroup
	cancel context.CancelFunc
}

func NewReloadCronLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ReloadCronLogic {
	return &ReloadCronLogic{
		ctx:    ctx,
		svcCtx: svcCtx,
		Logger: logx.WithContext(ctx),
	}
}

// 重载任务监听协程
func (l *ReloadCronLogic) ReloadCron(in *job.ReloadScheduleReq) (*job.ReloadScheduleRes, error) {
	l.mu.Lock()
	defer l.mu.Unlock()
	if l.cancel != nil {
		l.cancel()
		l.wg.Wait()
	}
	ctx, cancel := context.WithCancel(context.Background())
	l.cancel = cancel

	l.wg.Add(1)
	go func() {
		defer l.wg.Done()
		Consumer(ctx, l.svcCtx.Config)
	}()

	return &job.ReloadScheduleRes{}, nil
}

func Consumer(ctx context.Context, c config.Config) {
	c.MustSetUp()
	jobctx := context.Background()
	svcContext := svc.NewServiceContext(c)
	// 直接运行的程序
	logic.RunSimpleJob(jobctx, svcContext)
	waitTime := time.Second * 5
	if c.Mode == service.DevMode {
		waitTime = time.Millisecond * 300
	}
	time.AfterFunc(waitTime, func() {
		taskScheduler(ctx, svcContext, c)
	})
	taskJob(ctx, svcContext)
}

// 启动多个并发工作线程来处理客户端创建的任务
func taskJob(ctx context.Context, svcContext *svc.ServiceContext) {
	cronJob := logic.NewCronJob(ctx, svcContext)
	mux := cronJob.Register()

	logx.Infof("[task-job] starting server")

	// 检查取消信号
	go func() {
		<-ctx.Done()
		logx.Info("[task-job] received cancellation signal")
	}()

	if err := svcContext.AsynqServer.Run(mux); err != nil {
		logx.WithContext(ctx).Errorf("[task-job] run err : %+v", err)
	}
}

// 创建并安排由后台工作人员异步处理的任务
func taskScheduler(ctx context.Context, svcContext *svc.ServiceContext, c config.Config) {
	cronScheduler := scheduler.NewCronScheduler(ctx, svcContext, c.JobConf.New())
	cronScheduler.Register()

	logx.Infof("[task-scheduler] starting server")

	// 检查取消信号
	go func() {
		<-ctx.Done()
		logx.Info("[task-scheduler] received cancellation signal")
	}()

	if err := svcContext.AsynqScheduler.Run(); err != nil {
		logx.WithContext(ctx).Errorf("[task-scheduler] run err : %+v", err)
	}
}
